package com.whq.thrift.api.kafka.requestlog;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.whq.common.StringUtil;
import com.whq.thrift.api.common.spring.ReloadableConfManager;
import com.whq.thrift.api.kafka.requestlog.service.RequestLogService;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;

public class MainKafkaRequestLoginLog {
	private static final Logger LOGGER = LoggerFactory.getLogger(MainKafkaRequestLoginLog.class);
	
	private final ConsumerConnector consumer;
	private ReloadableConfManager reloadableConfManager = null;
	private RequestLogService requestLogService = null;
//	private IceClientHelper iceClientHelper = null;
	private String topic = "";
	private ThreadPoolExecutor executor = null;
	private ClassPathXmlApplicationContext ctx = null;

	public MainKafkaRequestLoginLog() {
		// 加载spring配置
		ctx = new ClassPathXmlApplicationContext("classpath:/conf/spring/applicationContext-core.xml");
		reloadableConfManager = (ReloadableConfManager) ctx.getBean("reloadableConfManager");
		requestLogService = ctx.getBean(RequestLogService.class);
//		iceClientHelper = (IceClientHelper) ctx.getBean("iceClientHelper");
		topic = reloadableConfManager.getPropertyValue("kafka.consumer.receiverequestlog.topic", null);
		
		Properties props = new Properties();
		// zookeeper 配置
		props.put("zookeeper.connect", reloadableConfManager.getPropertyValue("kafka.consumer.receiverequestlog.zookeeper.connect", null));

		// group 代表一个消费组
		props.put("group.id", reloadableConfManager.getPropertyValue("kafka.consumer.receiverequestlog.group.id", null));

		// zk连接超时
		props.put("zookeeper.session.timeout.ms", reloadableConfManager.getPropertyValue("kafka.consumer.receiverequestlog.zookeeper.session.timeout.ms", null));
		props.put("zookeeper.sync.time.ms", reloadableConfManager.getPropertyValue("kafka.consumer.receiverequestlog.zookeeper.sync.time.ms", null));
		props.put("auto.commit.interval.ms", reloadableConfManager.getPropertyValue("kafka.consumer.receiverequestlog.auto.commit.interval.ms", null));
		props.put("auto.offset.reset", reloadableConfManager.getPropertyValue("kafka.consumer.receiverequestlog.auto.offset.reset", null));
		// 序列化类
		props.put("serializer.class", "kafka.serializer.StringEncoder");

		ConsumerConfig config = new ConsumerConfig(props);

		consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
		
		// 初始化处理任务的线程池
		// 保留的线程池大小
		int corePoolSize = reloadableConfManager.getIntPropertyValue("saverequestlog.threadpool.concurrentThreadCount", null, 10);
		// 线程池的最大大小
		int maximumPoolSize = reloadableConfManager.getIntPropertyValue("saverequestlog.threadpool.threadQueueSize", null, 20);
		// 空闲线程结束的超时时间
		long keepAliveTime = reloadableConfManager.getIntPropertyValue("saverequestlog.threadpool.threadKeepAliveSeconds", null, 600);
		// keepAliveTime 的单位
		TimeUnit keepAliveTimeUnit = TimeUnit.SECONDS;
		
		// 存放任务的队列
		final BlockingQueue<Runnable> queueHandleTask = new LinkedBlockingQueue<Runnable>();
		executor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime,
				keepAliveTimeUnit, queueHandleTask);
	}

	public void consume() {
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, new Integer(1));

		StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
		StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());

		Map<String, List<KafkaStream<String, String>>> consumerMap = consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
		KafkaStream<String, String> stream = consumerMap.get(topic).get(0);
		ConsumerIterator<String, String> it = stream.iterator();
		
		long sleepMs = reloadableConfManager.getLongPropertyValue("sleep.ms.get.kafka.msg", null, 5);
		
		while (it.hasNext()) {
			try {
				Thread.sleep(sleepMs);
			} catch (Exception e) {
				LOGGER.error("Thread.sleep() exception: ", e);
			}
			
			String msg = it.next().message();
			if (StringUtil.isBlank(msg)) {
				continue;
			}
//			System.out.println(it.next().message());
			executor.execute(new ParseRequestLogMsgRunnable(msg, requestLogService, reloadableConfManager));
		}
	}
	
	public void closeSpringCtx() {
		if (ctx != null) {
			ctx.close();
		}
	}
	
	public static void main(String[] args) {
		MainKafkaRequestLoginLog kafkaConsumerLoginLog = new MainKafkaRequestLoginLog();
		kafkaConsumerLoginLog.consume();
		kafkaConsumerLoginLog.closeSpringCtx();
	}
}
